package com.etc.datefrme

import org.apache.spark.sql.SparkSession

/**
  * action操作详解
  *
  * collect、count、first、foreach、reduce、show、take
  *
  */
object Dataset {

  case class User(id : Long,name: String, age: Long)
  case class User1(uid : Long,name: String, age: Long)
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("Dataset")
      .master("local")
      .getOrCreate()
    import org.apache.spark.sql.functions._

    val frame = spark.read.json("e:\\a.json")
    val frame1 = spark.read.json("e:\\b.json")
    /*    //collect获取所有数据
    unit.collect().foreach(print)
    //count统计个数
    print(unit.count())
    // first：获取数据集中的第一条数据
    print(unit.first())
    // foreach：遍历数据集中的每一条数据，对数据进行操作，这个跟collect不同，collect是将数据获取到driver端进行操作
    // foreach是将计算操作推到集群上去分布式执行
    // foreach(println(_))这种，真正在集群中执行的时候，是没用的，因为输出的结果是在分布式的集群中的，我们是看不到的
    unit.foreach{println(_)}
    // reduce：对数据集中的所有数据进行归约的操作，多条变成一条
    // 用reduce来实现数据集的个数的统计
    println(unit.map(unit => 1).reduce(_+_))
    // show，默认将dataset数据打印前20条
    unit.show()
    // take，从数据集中获取指定条数
    unit.take(6).foreach(println)*/
    /*    // 持久化，在rdd部分仔细讲解过，我们这里就不展开讲了
    // 持久化，如果要对一个dataset重复计算两次的话，那么建议先对这个dataset进行持久化再进行操作，避免重复计算
    frame.cache()
    println(frame.count())
    println(frame.count())
    // 创建临时视图，主要是为了，可以直接对数据执行sql语句
    frame.createOrReplaceTempView("user")
    spark.sql("select * from user where age > 18").show()
    // dataframe/dataset，比如执行了一个sql语句获取的dataframe，实际上内部包含一个logical plan，逻辑执行计划
    // 设计执行的时候，首先会通过底层的catalyst optimizer，生成物理执行计划，比如说会做一些优化，比如push filter
    // 还会通过whole-stage code generation技术去自动化生成代码，提升执行性能
    frame.createOrReplaceTempView("user")
    spark.sql("select * from user where age > 18").explain()


    frame.printSchema()
    val unit = frame.as[User]
    unit.show()
    unit.printSchema()
    val frame1 = unit.toDF()*/
    //repartition增加分区
    //coalesce减少分区
    /*
    println(unit.rdd.partitions.size)
    val unit1 = unit.repartition(7)
    println(unit1.rdd.partitions.size)
    val unit3 = unit1.coalesce(3)
    println(unit3.rdd.partitions.size)
    unit3.show()*/

    /*    // distinct和dropDuplicates
    // 都是用来进行去重的，区别在哪儿呢？
    // distinct，是根据每一条数据，进行完整内容的比对和去重
    // dropDuplicates，可以根据指定的字段进行去重

//    val unit1 = unit.distinct()
//    unit1.show()
    val unit2 = unit.dropDuplicates("age")
    unit2.show()*/
    // except：获取在当前dataset中有，但是在另外一个dataset中没有的元素
    // filter：根据我们自己的逻辑，如果返回true，那么就保留该元素，否则就过滤掉该元素
    // intersect：获取两个数据集的交集
//    val unit = frame.as[User]
//    val unit1 = frame1.as[User1]
    //    unit.except(unit1).show()
    //    unit.filter(a => a.age>13).show()
    //    unit.intersect(unit1).show()

    // map：将数据集中的每条数据都做一个映射，返回一条新数据
    // flatMap：数据集中的每条数据都可以返回多条数据
    // mapPartitions：一次性对一个partition中的数据进行处理
    //    unit.map{a => (a.name,a.age + 10)}.show()
    //    unit.flatMap{
    //      user => Seq(User(user.name+ "_1",user.age + 1)
    //        ,User(user.name+ "_2",user.age + 2))
    //    }.show()
/*    unit.mapPartitions { user => {
      val result = scala.collection.mutable.ArrayBuffer[(String, Long)]()
      while (user.hasNext) {
        val emp = user.next()
        result += ((emp.name, emp.age + 1000))
      }
      result.iterator
      }
    }.show()*/
    //交集
//    unit.joinWith(unit1,$"id" === $"uid").show()
    //排序
//    unit.sort($"age".desc).show()
//    val array = unit.randomSplit(Array(3, 10, 20))
//    array.foreach{ds => ds.show() }
//    unit.sample(false,0.3).show()
//    frame.join(frame1,$"id" === $"uid").groupBy(frame1("name"))
//      .agg(avg(frame("age")),sum(frame("age")),max(frame("age")),min(frame("age")),count(frame("name"))).show()
//
//    frame.groupBy(frame("id"))
//      .agg(collect_set(frame("name")),collect_list(frame("name")))
//      .collect()
//      .foreach(println(_))
    frame
      .select(frame("name"), current_date(), current_timestamp(), rand(), round(frame("age"), 2), concat(frame("name"), frame("age")))
      .show()
  }
}
